package com.personas.analysic.analyserv.map;

import com.personas.analysic.analyserv.entity.YearBase;

import com.personas.analysic.analyserv.util.DateUtil;
import com.personas.analysic.analyserv.util.HbaseUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.RichMapFunction;


public class YearBaseMap extends RichMapFunction<String,YearBase>{

    @Override
    public YearBase map(String str) throws Exception {

        if (StringUtils.isBlank(str)){
            return null;
        }

        String[] split = str.split(",");

        String userId = split[0];
        String userName = split[1];
        String passWord = split[2];
        String sex = split[3];
        String phone = split[4];
        String email = split[5];
        String age = split[6];
        String regTime = split[7];
        String adress = split[8];
        String terminal = split[9];

        String yearInfo = DateUtil.dateFormat(age);


        HbaseUtil.putdata("userInfo",userId,"baseInfo","age",age);
        HbaseUtil.putdata("userInfo",userId,"baseInfo","yearInfo",yearInfo);

        YearBase yearBase = new YearBase();
        yearBase.setYear(yearInfo);
        String group = "yearBaseType =" + yearInfo;
        yearBase.setGroup(group);
        return yearBase;
    }
}
